package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Author: Felix
 * Date: 2021/8/6
 * Desc: 用户跳出明细统计
 */
public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        /*////2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //2.3 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //2.4 设置job取消后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.5 设置状态后端   内存|文件系统|RocksDB
        env.setStateBackend(new FsStateBackend("hdfs://hadoop202:8020/ck/gmall"));
        //2.6 指定操作HDFS的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 3.从kafka中读取数据
        //3.1 声明消费主题以及消费者组
        String topic = "dwd_page_log";
        String groupId = "user_jump_detail_app_group";
        //3.2 获取kafka消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 读取数据封装流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //测试数据
//        DataStream<String> kafkaDS = env
//            .fromElements(
//                "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
//                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":12000}",
//                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
//                    "\"home\"},\"ts\":15000} ",
//                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
//                    "\"detail\"},\"ts\":30000} "
//            );

        //TODO 4.对读取的数据进行类型转换 String ->JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
//        jsonObjDS.print(">>>>");

        //TODO 5.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> jsonObjWithWatermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
                //本次采用单调递增
                //制定策略 flink 分两种
                //一种单调递增 （forMonotonousTimestamps）
                //一种乱序 （forBoundedOutOfOrderness）
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts");//跳入时间
                                    }
                                }
                        )
        );

        //TODO 6.按照mid进行分组
        //common 公共信息
        //mid 设备id
        KeyedStream<JSONObject, String> keyedDS = jsonObjWithWatermarkDS.keyBy(
                jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //TODO 7.定义pattern（配置CEP表达式）
        //跳转:10秒钟之内的操作,跳转其它页面
        //跳出:首次访问之后很长一段时间（自己设定），用户没继续再有其他页面的访问,访问了一个页面，并且last_id不等于空(访客这次访问的第一个页面)
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first").where(
                //条件1：开启一个新的会话访问
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObj){
                        //获取上页的类型   page--lase_page_id 上页类型
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (lastPageId == null || lastPageId.length() == 0 ){
                            return true;
                        }
                        return false;
                    }
                }
        ).next("second").where(
                //条件2：访问了网站的其它页面
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObj){
                        //获取页面id   page--page_id 上页类型
                        String pageId = jsonObj.getJSONObject("page").getString("page_id");
                        if (pageId != null && pageId.length() > 0){
                            return true;
                        }
                        return false;
                    }
                }
        ).within(Time.seconds(10));

        //TODO 8.将pattern应用到流上（根据表达式筛选流）
        //CEP非常适合通过多条数据组合来识别某个事件
        PatternStream<JSONObject> patternDS = CEP.pattern(keyedDS, pattern);

        //TODO 9.从流中提取数据
        //9.1 定义侧输出流标记   FlinkCEP会将超时数据匹配放到侧输出流中
        OutputTag<String> timeoutTag = new OutputTag<String>("timeoutTag"){
        };
        //9.2 提取数据
        //超时跳出
        SingleOutputStreamOperator<String> resDS = patternDS.flatSelect(
                timeoutTag,
                //处理超时数据
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp, Collector<String> out) throws Exception {
                        //超时情况  就是我们要统计的跳出
                        List<JSONObject> jsonObjectList = pattern.get("first");
                        for (JSONObject jsonObj : jsonObjectList) {
                            out.collect(jsonObj.toJSONString());
                        }
                    }
                },
                //处理完全匹配数据的
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<String> out) throws Exception {
                        //完全匹配的数据  是跳转情况，不在我们的统计范围之内
                    }
                }
        );

        //9.3 从侧输出流中获取超时数据（跳出）
        DataStream<String> jumpDS = resDS.getSideOutput(timeoutTag);
        jumpDS.print(">>>>");

        //TODO 10.将跳出明细写到kafka的dwm层主题
        jumpDS.addSink(
                MyKafkaUtil.getKafkaSink("dwm_user_jump_detail")
        );

        env.execute();
    }
}

